package plgnet

import (
	"dgo/goutils/utils"
	"errors"
	"fmt"
	"gitee.com/ymofen/gobase/golog"
	"gitee.com/ymofen/supervalue"
	"golang.org/x/net/websocket"
	"net/http"
	"strings"
	"sync/atomic"
)

type WsServiceConn struct {
	sessionid  string
	remoteAddr string
	owner      *PlgHttpWsService
	conn       *websocket.Conn
	channelMap utils.GSyncMapGroup

	trans_attach_info bool
}

func (this *WsServiceConn) SendV(data interface{}) error {
	var err error
	var buf []byte
	if str, ok := data.(string); ok {
		buf = []byte(str)
	} else if json, ok := data.(*wrapper.SuperValue); ok {
		buf = json.AsJSONBytesEx(false, true)
	} else if b, ok := data.([]byte); ok {
		buf = make([]byte, len(b))
		copy(buf, b)
	} else {
		err = fmt.Errorf("未知的发送数据:%v", data)
	}
	if len(buf) > 0 { // 使用排队发送, 避免订阅多个通道出现发送乱序的情况
		err = this.owner.workers.PostTaskFunc(this.sessionid, func() {
			err := SendWsMessage(this.conn, WS_MESSAGE_TEXT, buf)
			if err != nil {
				golog.Warnf("发送Websocket数据异常:%s", err.Error())
			} else {
				this.owner.monitor_send.DeltaBuf(buf)
			}
		})
	}

	return err
}

func (this *WsServiceConn) SendKVS(kvs ...interface{}) error {
	l := len(kvs)
	if l == 0 {
		return errors.New("没有需要发送的数据")
	}
	if l == 1 {
		return this.SendV(kvs[0])
	}

	json := wrapper.NewSVObject()
	json.SetKeyValues(kvs...)
	return this.SendV(json)
}

func (this *WsServiceConn) GetSessionID() string {
	return this.sessionid
}

func (this *WsServiceConn) onRecvSendMessageFunc(sender interface{}, channelid string, subscribeid interface{}, data interface{}, tag interface{}, isBreak *bool) bool {
	this.SendV(data)
	return true
}

func (this *WsServiceConn) SubscribleChannel(channels string) int {
	r := 0
	strs := strings.Split(channels, ",")
	for _, str := range strs {
		channelid := utils.Trim(str)
		if len(channelid) == 0 {
			continue
		}
		itm := this.channelMap.Get(channelid, nil)
		if itm == nil {
			this.owner.ws_ex.SubscribeChannel(channelid, this, this.onRecvSendMessageFunc)
			this.channelMap.Set(channelid, this)
			r++
		}
	}
	return r
}

func (this *WsServiceConn) UnsubscribleChannel(channels string) int {
	r := 0
	strs := strings.Split(channels, ",")
	for _, str := range strs {
		channelid := utils.Trim(str)
		if len(channelid) == 0 {
			continue
		}
		itm := this.channelMap.Get(channelid, nil)
		if itm != nil {
			this.owner.ws_ex.UnsubscribeChannel(channelid, this)
			this.channelMap.Remove(channelid)
			r++
		}
	}
	return r
}

func (this *WsServiceConn) onClose() {
	owner := this.owner
	if owner != nil {
		this.channelMap.Range(func(key, value interface{}) bool {
			channelid := key.(string)
			owner.ws_ex.UnsubscribeChannel(channelid, this)
			return true
		})
	}
	this.conn = nil
}

func (this *WsServiceConn) Request() *http.Request {
	conn := this.conn
	if conn == nil {
		return nil
	}
	return this.conn.Request()
}

func (this *WsServiceConn) RequestDis() {
	conn := this.conn
	if conn == nil {
		return
	}

	this.SendKVS("code", -1, "msg", "token is invalid!")
	conn.Close()
}

func (this *PlgHttpWsService) onHttpHandle(conn *websocket.Conn) {

	if utils.GoFunCatchException {
		defer utils.PanicHandlerWithDebugString("WsHttpService-onHttpHandle")
	}

	wsconn := &WsServiceConn{owner: this, conn: conn, sessionid: utils.RandKeyString(24)}
	if conn.Request() != nil {
		wsconn.remoteAddr = conn.Request().RemoteAddr
	} else {
		wsconn.remoteAddr = conn.RemoteAddr().String()
	}
	this.ws_ex.SubscribeChannel(wsconn.sessionid, wsconn, wsconn.onRecvSendMessageFunc)
	this.ws_ex.DispatchData(WS_HTTP_CHANNEL_EVENT_ONCONNECTED, wsconn, "onconnect", nil)
	atomic.AddInt32(&this.online, 1)
	defer func() {
		atomic.AddInt32(&this.online, -1)
		this.ws_ex.UnsubscribeChannel(wsconn.sessionid, wsconn)
		wsconn.onClose()
	}()

	channels := wsconn.Request().URL.Query().Get("channels")
	if len(channels) > 0 {
		wsconn.SubscribleChannel(channels)
	}

	for {
		msgtype, msg, err := ReceiveMessage(conn)
		if err != nil {
			break
		}
		this.monitor_recv.DeltaBuf(msg)

		this.onRecv(wsconn, msgtype, msg)
	}

}

func (this *PlgHttpWsService) onRecvCmd(wsconn *WsServiceConn, json *wrapper.SuperValue) {
	cmd := json.StringByName("cmd", "")
	if cmd == WS_HTTP_CMD_subscribe {
		str := json.StringByName("channels", "")
		r := wsconn.SubscribleChannel(str)
		json.CheckExistsIntFunc("request-attach", func(val int64) {
			wsconn.trans_attach_info = val == 1
		})

		wsconn.SendKVS("code", 0, "channels", str, "n", r, "client-addr", wsconn.remoteAddr)
	} else if cmd == WS_HTTP_CMD_unsubscribe {
		str := json.StringByName("channels", "")
		r := wsconn.UnsubscribleChannel(str)
		wsconn.SendKVS("code", 0, "channels", str, "n", r)
	}

}

func (this *PlgHttpWsService) onRecv(wsconn *WsServiceConn, msgtype byte, msg []byte) {
	var strHex string
	defer utils.PanicHandlerWithDebugString(strHex)

	if msgtype == WS_MESSAGE_TEXT {
		strHex = utils.BufToHexStr(msg, 0, "")

		json, err := wrapper.NewSVFromBuf(msg)
		if err != nil {
			var strmsg string
			if len(msg) < 128 {
				strmsg = string(msg)
			} else {
				strmsg = string(msg[:127]) + "..."
			}
			golog.Warnf("收到非法的Websocket数据:%s, from:%s, err:%s", strmsg, wsconn.remoteAddr, err.Error())
			return
		}
		topic := json.StringByName(WS_TOPIC_KEY, "")
		if len(topic) == 0 {
			topic = "unkown"
		} else if topic == WS_HTTP_CHANNEL_CMD {
			this.onRecvCmd(wsconn, json)
			return
		}

		if topic == "__ECHO" {
			wsconn.SendV(json)
			return
		}

		this.ws_ex.DispatchData(topic, wsconn, json, nil)
	} else {
		golog.Warnf("收到非法的Websocket数据, msgtype:%d, from:%s", msgtype, wsconn.remoteAddr)
	}
}
